package cn.bone.kafka.kafka02;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducer {

    private static Logger logger = LoggerFactory.getLogger(CustomProducer.class);

    public static void main(String[] args) {
        sendWithCallback();
    }



    /**
     * 带回调函数的异步发送
     */
    public static void sendWithCallback(){
        // 0. 配置
        Properties properties = new Properties();
        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.5:3001,192.168.1.5:3002,192.168.1.5:3003");
        // 指定对应的key和value的序列化 key.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 指定自定义的分区规则
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "cn.bone.kafka.kafka02.MyPartitioner");

        // 缓冲区大小（默认值：32M）
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 批次大小（默认值：16k）
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms（默认值：0ms）
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 0);
        // 压缩（默认none，可配置gzip，snappy，lz4，zstd，通常snappy用的比较多）
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        // 设置ack（类型是String，默认值是all，可以取值"0","1","-1","all"），"all"和"-1"是相同的效果
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        // 设置重试次数（类型是int，默认值是int的最大值）
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        // 开启幂等性（类型为boolean，默认开启true）
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 1. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 2. 发送数据
        for(int i=0; i<10; i++){

            // 发送的值中，如果包含abc，消息会发到分区0，否则会发送到分区1
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first", "hello_callback_abc"+i);
            kafkaProducer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception == null){
                        logger.info("success! 主题: {}, 分区: {}", metadata.topic(), metadata.partition());
                    }
                }
            });
        }

        // 3. 关闭资源
        kafkaProducer.close();
    }

}
